package com.jch.learn.demo.code;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;

// 固定消息长度格式
public class BasicQueue {

    // 队列最多消息个数，实际个数减1
    private static final int MAX_MSG_NUM = 100;

    // 消息体最大长度
    private static final int MAX_MSG_BODY_SIZE = 1020;

    // 每条消息占用的空间 (长度 + 消息体长度)
    private static final int MSG_SIZE = MAX_MSG_BODY_SIZE + 4;

    // 队列消息体数据文件大小
    private static final int DATA_FILE_SIZE = MAX_MSG_NUM * MSG_SIZE;

    // 队列元数据文件大小 （head + tail）
    private static final int META_SIZE = 8;

    // 数据
    private MappedByteBuffer dataBuf;
    // 元数据 （head，tail）
    private MappedByteBuffer metaBuf;

    public BasicQueue(String path, String queueName) throws IOException {
        if (!path.endsWith(File.separator)) {
            path += File.separator;
        }
        RandomAccessFile dataFile = null;
        RandomAccessFile metaFile = null;
        try {
            dataFile = new RandomAccessFile(path + queueName + ".data", "rw");
            metaFile = new RandomAccessFile(path + queueName + ".meta", "rw");
            dataBuf = dataFile.getChannel().map(FileChannel.MapMode.READ_WRITE, 0, DATA_FILE_SIZE);
            metaBuf = metaFile.getChannel().map(FileChannel.MapMode.READ_WRITE, 0, META_SIZE);
        } finally {
            if (dataFile != null) {
                dataFile.close();
            }
            if (metaFile != null) {
                metaFile.close();
            }
        }
    }

    // 获取消息头位置
    private int head() {
        return metaBuf.getInt(0);
    }

    // 设置新的消息头位置
    private void head(int newHead) {
        metaBuf.putInt(0, newHead);
    }

    // 获取消息尾位置
    private int tail() {
        return metaBuf.getInt(4);
    }

    // 设置新的消息尾位置
    private void tail(int newTail) {
        metaBuf.putInt(4, newTail);
    }

    // 队列是否为空
    private boolean isEmpty() {
        return head() == tail();
    }

    // 队列是否已满
    private boolean isFull() {
        return (tail() + MSG_SIZE) % DATA_FILE_SIZE == head();
    }

    /**
     * 入队
     *
     * @param data 消息
     */
    public void enqueue(byte[] data) {
        if (data.length > MAX_MSG_BODY_SIZE) {
            throw new IllegalArgumentException("msg size is " + data.length + ",while maximum allowed length is " + MAX_MSG_BODY_SIZE);
        }
        if (isFull()) {
            throw new IllegalStateException("queue is full.");
        }
        // 找到队列尾，定位到队列尾，写消息长度，写实际数据。
        int tail = tail();
        dataBuf.position(tail);
        dataBuf.putInt(data.length);
        dataBuf.put(data);
        if (tail + MSG_SIZE >= DATA_FILE_SIZE) { // 更新队列尾指针，如果已到文件尾，再从头开始。
            tail(0);
        } else {
            tail(tail + MSG_SIZE);
        }
    }

    /**
     * 出队
     * @return byte[]
     */
    public byte[] dequeue() {
        if (isEmpty()) {
            return null;
        }
        int head = head();
        dataBuf.position(head);
        int length = dataBuf.getInt();
        byte[] data = new byte[length];
        dataBuf.get(data);
        if (head + MSG_SIZE >= DATA_FILE_SIZE) {
            head(0);
        } else {
            head(head + MSG_SIZE);
        }
        return data;
    }
}
